package com.cet.ej.job;

import com.cet.ej.domain.Foo;
import org.apache.shardingsphere.elasticjob.api.ShardingContext;
import org.apache.shardingsphere.elasticjob.dataflow.job.DataflowJob;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;

/**
 * 数据流作业
 * @author 30952
 */

@Component
public class MyDataFlowJob implements DataflowJob<Foo> {
    @Override
    public List<Foo> fetchData(ShardingContext shardingContext) {
        // streaming.process 配置是否开启流式处理，不开启则只调用一次fetchData和processData
        // 在方法fetchData返回null 或者 空集合则停止抓取

        // shardingContext::taskId  dataflowJob@-@0,1,2@-@READY@-@172.22.128.1@-@11696
        List<Foo> data = new ArrayList<>();
        switch (shardingContext.getShardingItem()) {
            case 0:
                data = getData();// get data from database by sharding item 0
                return data;
            case 1:
                 data = getData();// get data from database by sharding item 1
                return data;
            case 2:
                 data = getData();// get data from database by sharding item 2
                return data;
            // case n: ...
        }
        return data;
    }

    @Override
    public void processData(ShardingContext shardingContext, List list) {
        // process data
        // 处理完数据建议更新状态，否则在流式处理中会无止境地调用fetchData方法
    }

    private List<Foo> getData() {
        return Collections.emptyList();
    }
}
